package com.imooc.web.lock;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    static final Logger log = LoggerFactory.getLogger(DistributedLock.class);

    /**
     * zk客户端
     */
    private CuratorFramework client;

    /**
     * 用于挂起未抢到锁的进程
     */
    private static CountDownLatch zkLockLatch = new CountDownLatch(1);

    /**
     * 所有业务的分布式锁都在该节点下，用于管理本项目使用的所有锁
     */
    private static final String ZK_LOCK_PROJECT = "zk-locks";

    /**
     * 本次业务使用的分布式锁
     */
    private static final String ZK_DISTRIBUTED_LOCK = "distributed_lock";

    /**
     * 获取spring容器中初始化的zk客户端
     * @param client zk客户端
     */
    public DistributedLock(CuratorFramework client) {//可以通过spring容器注入zk客户端实例
        this.client = client;
    }

    /**
     * 初始化锁
     */
    public void init(){
        client = client.usingNamespace("ZKLocks-Namespace");
        try {
            if(client.checkExists().forPath("/"+ZK_LOCK_PROJECT) == null){
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)//curator4.0规定临时节点下边不能创建子节点，所以父节点要是持久节点
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/"+ZK_LOCK_PROJECT);
            }
            // 针对zk的分布式节点，创建相应的watcher事件监听
            // 由于采用zk客户端是curator，所以监听的是父节点，若是zk的原生客户端，则监听ZK_DISTRIBUTED_LOCK
            addWatcherForLock("/"+ZK_LOCK_PROJECT);
        } catch (Exception exception) {
            log.error("客户端连接zookeeper服务器错误...请重试...");
        }
    }

    /**
     * 为节点设置监听事件
     * @param path zk节点的路径
     */
    private void addWatcherForLock(String path) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client,path,true);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);//初始化节点数据到cache（初始化模式随便选择，不是重点）
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                    String path = event.getData().getPath();
                    if(path.contains(ZK_DISTRIBUTED_LOCK)){
                        log.info("watch：上一个会话已经释放了锁或该会话已断开，节点路径为："+path);
                        // 释放当前线程的计数器，让当前线程尝试获取锁
                        zkLockLatch.countDown();
                    }
                }
            }
        });

    }

    /**
     * 获取分布式锁
     */
    public void getLock(){
        while (true){//当上一个进程释放锁，且当前请求获取到锁才会跳出循环
            try {
                //获取锁的过程实际上就是创建节点
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath("/"+ZK_LOCK_PROJECT+"/"+ZK_DISTRIBUTED_LOCK);
                log.info("线程"+Thread.currentThread().getId()+"获取分布式锁成功");
                return;
            } catch (Exception exception) {
                log.error("线程"+Thread.currentThread().getId()+"获取分布式锁失败");
//                exception.printStackTrace();//调试的时候看日志使用
                try {
                    //若获取锁失败，将当前线程挂起
                    if(zkLockLatch.getCount() <= 0){
                        zkLockLatch = new CountDownLatch(1);//需要重新设置同步资源（因为可能之前调用过，计数已经为0）
                    }
                    // 阻塞线程
                    zkLockLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 释放锁
     */
    public void releaseLock(){
        try {
            if(client.checkExists().forPath("/"+ZK_LOCK_PROJECT+"/"+ZK_DISTRIBUTED_LOCK) != null){
                client.delete().forPath("/"+ZK_LOCK_PROJECT+"/"+ZK_DISTRIBUTED_LOCK);
            }
            log.info("线程"+Thread.currentThread().getId()+"释放锁成功");
        } catch (Exception exception) {
            exception.printStackTrace();
        }
    }

}
